package com.ljh.order.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author grung
 * @Date 2021/2/4 15:27
 * @Version 1.0
 */

@Configuration
@EnableRabbit
public class RabbitConfig {

    @Bean(name="rabbitMessageListenerContainer")
    public DirectMessageListenerContainer listenerContainer(ConnectionFactory connectionFactory){
        //写的时候，默认使用DEFAULT_NUM_THREADS =Runtime.getRuntime().availableProcessors() * 2个线程
        DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);

        //设置为手动接受消息模式
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setPrefetchCount(5);

        // 并发由setQueues和setConsumersPerQueue控制
        // 从哪些队列消费消息，就设置哪些队列
        // container.setQueues(orderQueue(), ttlQueue(), dlxQueue());
        // container.setQueueNames("", "", "", "");
        // 设置每个队列由几个消费者消费。
        // 如果容器已经在运行，该值可以动态设置。
        // 每个消费者使用一个Channel
        container.setConsumersPerQueue(5);

        // 一次确认几个消息
        container.setMessagesPerAck(1);
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(10);
        taskExecutor.setMaxPoolSize(20);
        // 设置该属性，灵活设置并发
        container.setTaskExecutor(taskExecutor);

        return container;
    }

    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    /**
     * 下订单正常队列
     * @return
     */
    @Bean
    public Queue orderQueue(){

        return QueueBuilder.durable("orderqueue.go").build();
    }

    /**
     * 订单待付款队列（延时队列）
     * @return
     */
    @Bean
    public Queue delayQueue(){
        Map<String, Object> props = new HashMap<>();
        //消息的生存时间10s
        props.put("x-message-ttl",10000);
        // 设置该队列所关联的死信交换器（当队列消息TTL到期后依然没有消费，则加入死信队列）
        props.put("x-dead-letter-exchange","order.dlx");
        // 设置该队列所关联的死信交换器的routingKey，如果没有特殊指定，使用原队列的routingKey
        props.put("x-dead-letter-routing-key", "k.dlx");
        Queue queue = new Queue("orderqueue.ttl", true, false, false, props);
        return queue;
    }



    /**
     * 订单取消队列（死信队列）
     * @return
     */
    @Bean
    public Queue queueDlx() {
        Queue queue = new Queue("orderqueue.dlx", true, false, false,null);
        return queue;
    }


    /**
     * 订单交换器
     * @return
     */
    @Bean
    public Exchange orderExchange() {
        DirectExchange exchange = new DirectExchange("order.go", true,
                false, null);
        return exchange;
    }


    /**
     * 延迟交换器
     * @return
     */
    public Exchange ttlExchange(){
        Map<String, Object> args = new HashMap<>();
        return new DirectExchange("order.ttl", true, false, args);
    }

    /**
     * 死信交换器
     * @return
     */
    @Bean
    public Exchange exchangeDlx() {
        DirectExchange exchange = new DirectExchange("order.dlx", true,
                false, null);
        return exchange;
    }

    /**
     * 用于发送订单
     * @return
     */
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with("k.go").noargs();
    }

    /**
     * 用于等待支付延迟队列
     * @return
     */
    @Bean
    public Binding ttlBinding() {
        return BindingBuilder.bind(delayQueue()).to(ttlExchange()).with("k.ttl").noargs();
    }

    /**
     * 死信交换器绑定死信队列
     * @return
     */
    @Bean
    public Binding bindingDlx() {
        return BindingBuilder.bind(queueDlx()).to(exchangeDlx()).with("k.dlx").noargs();
    }
}
